use std::env;
use std::io::{Read, Write};
use std::net::TcpListener;

use chrono::{DateTime, Local, NaiveDateTime};
use crypto::digest::Digest;
use crypto::md5::Md5;
use dotenv::dotenv;
use serde::{Deserialize, Serialize};
use sqlx::{PgPool, Pool, Postgres};
use sqlx::postgres::PgPoolOptions;

#[actix_rt::main]
async fn main() -> Result<(), sqlx::Error> {
    dotenv().ok();
    let listener = TcpListener::bind(":::8081").unwrap();
    let connection_str = env::var("DATABASE_URL")
        .expect("数据库连接字符串获取失败，请检查env文件是否已配置数据库连接字符串");
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(&connection_str).await.unwrap();
    for stream in listener.incoming() {
        let mut stream = stream.unwrap();
        let mut buffer = [0; 1024];
        stream.read(&mut buffer).unwrap();
        let read_content = std::str::from_utf8_mut(&mut buffer).unwrap().trim_matches('\u{0}');
        let dto: MonitorAcceptInfoDTO = serde_json::from_str(&read_content).unwrap();
        let cm = dto.command.unwrap();
        let resp = if cm == 35 {
            save_monitor_type(&dto, &pool).await
        } else if cm == 42 {
            save_monitor_data(&dto, &pool).await
        } else if cm == 43 {
            save_monitor_raw_data(&dto, &pool).await
        } else if cm == 64 {
            heart_dance(&dto, &pool).await
        } else {
            MonitorResponse::new(500, "commnad is illeage".to_string(), MonitorAcceptDataVO::new(cm.to_string(), dto.device_id.unwrap()))
        };
        let write_content = serde_json::to_string(&resp).unwrap();
        stream.write(write_content.as_bytes()).unwrap();
    }
    println!("Running on port 3000...");
    Ok(())
}


async fn save_monitor_type(dto: &MonitorAcceptInfoDTO, pool: &Pool<Postgres>) -> MonitorResponse<MonitorAcceptDataVO> {
    let user_id = &dto.user.clone().unwrap().to_uppercase();
    ;
    let cypher = &dto.cypher.clone().unwrap();
    let command = &dto.command.unwrap();
    let device_id = dto.device_id.clone().unwrap();
    let mut md5 = Md5::new();
    md5.input_str(&cypher);
    // 获取md5值
    let en_cypher = md5.result_str();
    let bim_user_op = get_user(&pool, &user_id).await.unwrap();
    let bim_user = bim_user_op.expect("this user is not exists");
    let vo = MonitorAcceptDataVO::new(command.to_string(), device_id.to_string());
    if !bim_user.cypher.eq(en_cypher.as_str()) {
        return build_fail(vo, "password error".to_string()).await;
    }
    let sensor_type = &dto.sensor_type.clone().unwrap();
    let org_code = bim_user.org_code.clone().to_string();
    let monitor_type_op = sqlx::query!(r#"select id,command,monitor_name,type_code,org_code from bim_monitor_device where id=$1 "#, &device_id)
        .fetch_optional(&*pool).await.unwrap();
    let moitor_type: BimMonitorType;
    if monitor_type_op.is_none() {
        // 新增
        let result = sqlx::query::<Postgres>(r#"insert into bim_monitor_device(id,command,monitor_name,type_code,org_code) values($1,$2,$3,$4,$5) "#)
            .bind(&device_id)
            .bind(command)
            .bind(&device_id)
            .bind(&sensor_type)
            .bind(&org_code)
            .execute(&*pool).await;
        match result {
            Ok(pqr) => {
                println!("{:?}", pqr);
            }
            Err(err) => {
                println!("err = {:?}", err);
            }
        }
    }

    let monitor_device_type_op = sqlx::query!("select org_code,type_code from bim_monitor_device_type where  org_code= $1 and type_code=$2",&org_code,&sensor_type)
        .fetch_optional(&*pool).await.unwrap();
    if monitor_device_type_op.is_none() {
        let now: DateTime<Local> = Local::now();
        // 新增
        let result = sqlx::query::<Postgres>(r#"insert into bim_monitor_device_type(type_code,org_code,created_name,created_date) values($1,$2,$3,$4) "#)
            .bind(&sensor_type)
            .bind(&org_code)
            .bind(&user_id)
            .bind(&now)
            .execute(&*pool).await;
        match result {
            Ok(pqr) => {
                println!("{:?}", pqr);
            }
            Err(err) => {
                println!("err = {:?}", err);
            }
        }
    }

    let response = build_success(vo).await;
    response
}

async fn save_monitor_data(dto: &MonitorAcceptInfoDTO, pool: &Pool<Postgres>) -> MonitorResponse<MonitorAcceptDataVO> {
    let user_id = &dto.user.clone().unwrap().to_uppercase();
    ;
    let cypher = &dto.cypher.clone().unwrap();
    let command = &dto.command.unwrap();
    let device_id = &dto.device_id.clone().unwrap();
    let theta = &dto.theta.clone().unwrap();
    let phi = &dto.phi.clone().unwrap();
    let psi = &dto.psi.clone().unwrap();
    let temperature = &dto.temperature.clone().unwrap();
    let device_val: f64 = temperature.parse::<f64>().unwrap();
    let mut md5 = Md5::new();
    md5.input_str(&cypher);
    // 获取md5值
    let en_cypher = md5.result_str();
    let bim_user_op = get_user(&pool, &user_id).await.unwrap();
    let bim_user = bim_user_op.expect("this user is not exists");
    let vo = MonitorAcceptDataVO::new(command.to_string(), device_id.to_string());
    if !bim_user.cypher.eq(en_cypher.as_str()) {
        return build_fail(vo, "password error".to_string()).await;
    }
    let org_code = bim_user.org_code.clone();
    // sqlx::query!(" select count(*) from bim_monitor_device where where id =? and org_code=? ",&device_id,&org_code).fetch_optional(&*pool).await.unwrap();
    let device = sqlx::query!("select type_code from bim_monitor_device where  org_code= $1 and id=$2",&org_code,&device_id)
        .fetch_optional(&*pool).await.unwrap();
    if device.is_none() {
        return build_fail(vo, "this device is not exists".to_string()).await;
    }
    let result = sqlx::query::<Postgres>(r#"insert into bim_monitor_data(device_id,command,theta,phi,psi,device_val,org_code,type_code,created_name,created_date) values($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)"#)
        .bind(&device_id)
        .bind(command.to_string())
        .bind(theta)
        .bind(phi)
        .bind(psi)
        .bind(device_val)
        .bind(&org_code)
        .bind(&device.unwrap().type_code)
        .bind(&user_id)
        .bind(Local::now())
        .execute(&*pool).await;
    let response = return match result {
        Ok(pqr) => {
            build_success(vo).await
        }
        Err(err) => {
            build_fail(vo, err.to_string()).await
        }
    };
    response
}

async fn save_monitor_raw_data(dto: &MonitorAcceptInfoDTO, pool: &Pool<Postgres>) -> MonitorResponse<MonitorAcceptDataVO> {
    let command = &dto.command.unwrap();
    let device_id = &dto.device_id.clone().unwrap();
    let x_value = &dto.x_value.clone().unwrap().parse::<i32>().unwrap();
    let y_value = &dto.y_value.clone().unwrap().parse::<i32>().unwrap();
    let z_value = &dto.z_value.clone().unwrap().parse::<i32>().unwrap();
    let user_id = &dto.user.clone().unwrap().to_uppercase();
    let cypher = &dto.cypher.clone().unwrap();
    let mut md5 = Md5::new();
    md5.input_str(&cypher);
    // 获取md5值
    let en_cypher = md5.result_str();
    let bim_user_op = get_user(&pool, &user_id).await.unwrap();
    let bim_user = bim_user_op.expect("this user is not exists");
    if !bim_user.cypher.eq(en_cypher.as_str()) {
        return build_fail(MonitorAcceptDataVO::new(command.to_string(), "".to_string()), "password error".to_string()).await;
    }
    let org_code = bim_user.org_code.clone();
    // sqlx::query!(" select count(*) from bim_monitor_device where where id =? and org_code=? ",&device_id,&org_code).fetch_optional(&*pool).await.unwrap();
    let device = sqlx::query!("select type_code from bim_monitor_device where  org_code= $1 and id=$2",&org_code,&device_id)
        .fetch_optional(&*pool).await.unwrap();
    if device.is_none() {
        return build_fail(MonitorAcceptDataVO::new(command.to_string(), device_id.to_string()), "this device is not exists".to_string()).await;
    }
    let type_code = &device.unwrap().type_code;
    let result = sqlx::query::<Postgres>(r#"insert into bim_monitor_raw_data(device_id,x_value,y_value,z_value,org_code,type_code,created_name,created_date) values($1,$2,$3,$4,$5,$6,$7,$8)"#)
        .bind(&device_id)
        .bind(x_value)
        .bind(y_value)
        .bind(z_value)
        .bind(&org_code)
        .bind(&type_code)
        .bind(&user_id)
        .bind(Local::now())
        .execute(&*pool).await;
    let flag = match result {
        Ok(pqr) => {
            println!("insert into bim_monitor_raw_data success {:?}", pqr);
            true
        }
        Err(err) => {
            println!("insert into bim_monitor_raw_data error {:?}", err);
            false
        }
    };
    if flag {
        let device_val: f64 = 0.0;
        let insert_result = sqlx::query::<Postgres>(r#"insert into bim_monitor_data(device_id,command,theta,phi,psi,device_val,org_code,type_code,created_name,created_date) values($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)"#)
            .bind(&device_id)
            .bind(command.to_string())
            .bind(x_value)
            .bind(y_value)
            .bind(z_value)
            .bind(device_val)
            .bind(&org_code)
            .bind(&type_code)
            .bind(&user_id)
            .bind(Local::now())
            .execute(&*pool).await;
        return match insert_result {
            Ok(pqr) => {
                println!("insert into bim_monitor_data success {:?}", pqr);
                build_success(MonitorAcceptDataVO::new(command.to_string(), dto.device_id.clone().unwrap())).await
            }
            Err(err) => {
                println!("insert into bim_monitor_data error {:?}", err);
                build_fail(MonitorAcceptDataVO::new(command.to_string(), dto.device_id.clone().unwrap()), err.to_string()).await
            }
        };
    } else {
        return build_fail(MonitorAcceptDataVO::new(command.to_string(), device_id.to_string()), "insert raw data failed".to_string()).await;
    }
}

async fn heart_dance(dto: &MonitorAcceptInfoDTO, pool: &Pool<Postgres>) -> MonitorResponse<MonitorAcceptDataVO> {
    let command = &dto.command.unwrap();
    let gateway_id = &dto.gateway_id.clone().unwrap();
    let user_id = &dto.user.clone().unwrap().to_uppercase();
    ;
    let cypher = &dto.cypher.clone().unwrap();
    let mut md5 = Md5::new();
    md5.input_str(&cypher);
    // 获取md5值
    let en_cypher = md5.result_str();
    let bim_user_op = get_user(&pool, &user_id).await.unwrap();
    let bim_user = bim_user_op.expect("this user is not exists");
    if !bim_user.cypher.eq(en_cypher.as_str()) {
        return build_fail(MonitorAcceptDataVO::new(command.to_string(), "".to_string()), "password error".to_string()).await;
    }
    let org_code = &bim_user.org_code;
    let gateway_op = sqlx::query_as::<_, BimMonitorGateway>(r#"select gateway_id ,device_ids  ,org_code,count from bim_monitor_gateway where gateway_id =$1 and org_code =$2"#)
        .bind(&gateway_id)
        .bind(&org_code)
        .fetch_optional(&*pool).await.unwrap();
    if gateway_op.is_none() {
        return build_fail(MonitorAcceptDataVO::new(command.to_string(), "".to_string()), "this gateway is not exist".to_string()).await;
    }
    let gateway = gateway_op.unwrap();
    let device_ids = gateway.device_ids.as_array().unwrap();
    let mut count = gateway.count;
    let mut cn = 0;
    let mut flag: bool = true;
    loop {
        if flag && cn < 3 {
            let one: i64 = 1;
            cn = cn + 1;
            let update_count = count + one;
            // println!("update_count:{}", update_count);
            let result = sqlx::query("update bim_monitor_gateway set  count = $1 , updated_date = $2 , updated_name = $3 where  gateway_id = $4 and org_code = $5 and count = $6")
                .bind(update_count)
                .bind(Local::now())
                .bind(user_id)
                .bind(gateway_id)
                .bind(org_code)
                .bind(count)
                .execute(&*pool).await;
            match result {
                Ok(pqr) => {
                    println!("{:?}", pqr);
                    flag = false;
                }
                Err(err) => {
                    let mg_op = sqlx::query_as::<_, BimMonitorGateway>(r#"select gateway_id ,device_ids  ,org_code,count from bim_monitor_gateway where gateway_id =$1 and org_code =$2"#)
                        .bind(&gateway_id)
                        .bind(&org_code)
                        .fetch_optional(&*pool).await.unwrap();
                    count = mg_op.unwrap().count;
                    println!("err = {:?}", err);
                }
            }
        } else {
            break;
        }
    }
    if flag {
        return build_fail(MonitorAcceptDataVO::new(command.to_string(), "".to_string()), "get device_id fail".to_string()).await;
    }
    let mut device_id_vec = Vec::new();
    for x in device_ids {
        device_id_vec.push(x.as_str().unwrap());
    }
    let len: i64 = device_ids.len() as i64;
    let index: usize = (count % len) as usize;
    let device_id = device_id_vec.get(index).unwrap();
    let vo = MonitorAcceptDataVO::new(command.to_string(), device_id.to_string());
    let response = build_success(vo).await;
    response
}

async fn build_fail(vo: MonitorAcceptDataVO, msg: String) -> MonitorResponse<MonitorAcceptDataVO> {
    MonitorResponse::new(500, msg, vo)
}

async fn build_success(vo: MonitorAcceptDataVO) -> MonitorResponse<MonitorAcceptDataVO> {
    let code = 200;
    let msg = "success".to_string();
    MonitorResponse::new(code, msg, vo)
}

async fn get_user(pool: &PgPool, user_id: &str) -> Result<Option<BimUser>, sqlx::Error> {
    let bim_user = sqlx::query_as!(
        BimUser,
        r#"SELECT id as user_id,cypher,org_code from bim_user where id=$1"#,
        user_id
    ).fetch_optional(pool)
        .await.unwrap();
    Ok(bim_user)
}

#[derive(Serialize, Deserialize, Debug, Default, Clone)]
#[serde_with::skip_serializing_none]
struct MonitorAcceptInfoDTO {
    user: Option<String>,
    cypher: Option<String>,
    device_id: Option<String>,
    command: Option<i32>,
    theta: Option<String>,
    phi: Option<String>,
    psi: Option<String>,
    temperature: Option<String>,
    sensor_type: Option<String>,
    gateway_id: Option<String>,
    x_value: Option<String>,
    y_value: Option<String>,
    z_value: Option<String>,
}

#[derive(Serialize, Deserialize, Debug, Default, Clone)]
#[serde_with::skip_serializing_none]
struct MonitorResponse<T> {
    code: i32,
    msg: String,
    data: T,
}

impl<T> MonitorResponse<T> {
    fn new(code: i32, msg: String, data: T) -> MonitorResponse<T> {
        MonitorResponse {
            code,
            msg,
            data,
        }
    }
}

#[derive(Serialize, Deserialize, Debug)]
struct MonitorAcceptDataVO {
    command: String,
    device_id: String,
}

impl MonitorAcceptDataVO {
    fn new(command: String, device_id: String) -> MonitorAcceptDataVO {
        MonitorAcceptDataVO {
            command,
            device_id,
        }
    }
}


#[derive(Serialize, Deserialize, Debug, Clone, Default, sqlx::FromRow)]
pub struct BimUser {
    pub user_id: String,
    pub cypher: String,
    pub org_code: String,

}

#[derive(Debug)]
pub struct BimMonitorType {
    pub id: String,
    pub command: String,
    pub monitor_name: String,
    pub type_code: String,
    pub org_code: String,
    pub updated_date: NaiveDateTime,
    pub updated_name: String,
    pub created_date: NaiveDateTime,
    pub created_name: String,

}

#[derive(Serialize, Deserialize, Debug, Clone, Default, sqlx::FromRow)]
pub struct BimMonitorGateway {
    pub gateway_id: String,
    pub device_ids: serde_json::Value,
    pub org_code: String,
    pub count: i64,

}